package com.example.toy.engine.spark;

import com.example.toy.core.dataset.BigCollection;
import com.example.toy.core.engine.EngineContext;
import com.example.toy.engine.dataset.SparkBigCollection;
import java.util.List;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;

/**
 * @author shenb
 * @date 2021-12-20 21:13
 */
public class SparkEngineContext implements EngineContext {

  private final SparkSession spark;
  private final JavaSparkContext jsc;

  public SparkEngineContext() {
    this.spark = SparkSession.builder().appName("toy").getOrCreate();
    this.jsc = new JavaSparkContext(spark.sparkContext());
  }

  @Override
  public <T> BigCollection<T> create(List<T> list) {

    JavaRDD<T> parallelize = jsc.parallelize(list);
    return new SparkBigCollection<>(this, parallelize);
  }

  @Override
  public <T> BigCollection<T> create(T... list) {
    return null;
  }

  @Override
  public <T> BigCollection<T> empty() {
    return new SparkBigCollection<>(this, jsc.emptyRDD());
  }
}
